/*
 * Copyright (c) 2015-2018 Sfkj Science And Technology Co.,Ltd.
 * All Rights Reserved.
 * This software is the confidential and proprietary information of
 * Sfkj Science And Technology Co.,Ltd.
 * ("Confidential Information").You shall not
 * disclose such Confidential Information and shall use it only in
 * accordance with the terms of the license agreement you entered into
 * with Sfkj.
 */
package com.sfkj.walk.axon.config;

import com.fasterxml.jackson.databind.DeserializationFeature;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.SerializationFeature;
import com.mongodb.MongoClient;
import org.axonframework.commandhandling.AsynchronousCommandBus;
import org.axonframework.commandhandling.CommandBus;
import org.axonframework.commandhandling.CommandMessage;
import org.axonframework.common.transaction.NoTransactionManager;
import org.axonframework.common.transaction.TransactionManager;
import org.axonframework.config.Configurer;
import org.axonframework.eventhandling.EventMessage;
import org.axonframework.eventhandling.MetaDataConstants;
import org.axonframework.eventhandling.interceptors.EventLoggingInterceptor;
import org.axonframework.eventsourcing.eventstore.EmbeddedEventStore;
import org.axonframework.eventsourcing.eventstore.EventStorageEngine;
import org.axonframework.eventsourcing.eventstore.EventStore;
import org.axonframework.messaging.MessageDispatchInterceptor;
import org.axonframework.messaging.MessageHandlerInterceptor;
import org.axonframework.messaging.correlation.SimpleCorrelationDataProvider;
import org.axonframework.messaging.interceptors.BeanValidationInterceptor;
import org.axonframework.messaging.interceptors.CorrelationDataInterceptor;
import org.axonframework.messaging.interceptors.LoggingInterceptor;
import org.axonframework.mongo.DefaultMongoTemplate;
import org.axonframework.mongo.MongoTemplate;
import org.axonframework.queryhandling.QueryBus;
import org.axonframework.queryhandling.SimpleQueryBus;
import org.axonframework.serialization.Serializer;
import org.axonframework.serialization.json.JacksonSerializer;
import org.axonframework.serialization.xml.XStreamSerializer;
import org.springframework.beans.factory.BeanClassLoaderAware;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;

import java.util.List;
import java.util.concurrent.Executors;

@Configuration
public class AxonConfiguration implements BeanClassLoaderAware {

    private ClassLoader beanClassLoader;
    private final Configurer configurer;

    public AxonConfiguration(Configurer configurer) {
        this.configurer = configurer;
    }

    @Bean
    public MongoTemplate axonMongoTemplate(MongoClient mongo, @Value("${axon.database}") String databaseName) {
        return new DefaultMongoTemplate(mongo, databaseName);
    }

    @Bean
    @Qualifier("tokenSerializer")
    public XStreamSerializer xStreamSerializer() {
        XStreamSerializer xStreamSerializer = new XStreamSerializer();
        xStreamSerializer.getXStream().setClassLoader(beanClassLoader);
        return xStreamSerializer;
    }

    @Primary
    @Bean
    public Serializer axonJacksonSerializer() {
        ObjectMapper objectMapper = new ObjectMapper();
        objectMapper.configure(SerializationFeature.WRITE_DATES_AS_TIMESTAMPS, false);
        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);

        return new JacksonSerializer(objectMapper);
    }

    @Bean
    public LoggingInterceptor loggingInterceptor() {
        return new LoggingInterceptor();
    }

    @Bean
    public BeanValidationInterceptor beanValidationInterceptor() {
        return new BeanValidationInterceptor();
    }

    @Bean
    public SimpleCorrelationDataProvider simpleCorrelationDataProvider() {
        return new SimpleCorrelationDataProvider(
                MetaDataConstants.CURRENT_USER_ID,
                MetaDataConstants.USER_AGENT,
                MetaDataConstants.HOST,
                MetaDataConstants.PORT,
                MetaDataConstants.COMMAND_ID
        );
    }

    @Primary
    @Bean(destroyMethod = "shutdown")
    public CommandBus commandBus(List<MessageHandlerInterceptor<CommandMessage<?>>> handlerInterceptors,
                                 List<MessageDispatchInterceptor<CommandMessage<?>>> dispatchInterceptors,
                                 org.axonframework.config.Configuration config,
                                 org.axonframework.spring.config.AxonConfiguration configuration) {
        AsynchronousCommandBus commandBus = new AsynchronousCommandBus(
                Executors.newCachedThreadPool(),
                config.getComponent(TransactionManager.class, () -> NoTransactionManager.INSTANCE),
                config.messageMonitor(CommandBus.class, "commandBus")
        );
        commandBus.registerHandlerInterceptor(new CorrelationDataInterceptor<>(configuration.correlationDataProviders()));
        for (MessageHandlerInterceptor handlerInterceptor : handlerInterceptors) {
            commandBus.registerHandlerInterceptor(handlerInterceptor);
        }
        for (MessageDispatchInterceptor dispatchInterceptor : dispatchInterceptors) {
            commandBus.registerDispatchInterceptor(dispatchInterceptor);
        }
        return commandBus;
    }

    @Bean
    public EventLoggingInterceptor eventLoggingInterceptor() {
        return new EventLoggingInterceptor();
    }

    @Qualifier("eventStore")
    @Bean(name = "eventBus")
    public EmbeddedEventStore eventStore(List<MessageDispatchInterceptor<EventMessage<?>>> dispatchInterceptors,
                                         EventStorageEngine storageEngine, org.axonframework.spring.config.AxonConfiguration configuration) {
        EmbeddedEventStore eventStore = new EmbeddedEventStore(storageEngine, configuration.messageMonitor(EventStore.class, "eventStore"));
        for (MessageDispatchInterceptor dispatchInterceptor : dispatchInterceptors) {
            eventStore.registerDispatchInterceptor(dispatchInterceptor);
        }
        return eventStore;
    }

    @Primary
    @Bean
    public QueryBus queryBus(List<MessageHandlerInterceptor<CommandMessage<?>>> handlerInterceptors,
                             List<MessageDispatchInterceptor<CommandMessage<?>>> dispatchInterceptors,
                             org.axonframework.config.Configuration config,
                             org.axonframework.spring.config.AxonConfiguration configuration) {
        SimpleQueryBus queryBus = new SimpleQueryBus(
                config.messageMonitor(CommandBus.class, "queryBus"),
                config.getComponent(TransactionManager.class, () -> NoTransactionManager.INSTANCE),
                null
        );
        queryBus.registerHandlerInterceptor(new CorrelationDataInterceptor<>(configuration.correlationDataProviders()));
        for (MessageHandlerInterceptor handlerInterceptor : handlerInterceptors) {
            queryBus.registerHandlerInterceptor(handlerInterceptor);
        }
        for (MessageDispatchInterceptor dispatchInterceptor : dispatchInterceptors) {
            queryBus.registerDispatchInterceptor(dispatchInterceptor);
        }
        return queryBus;
    }

    @Override
    public void setBeanClassLoader(ClassLoader classLoader) {
        this.beanClassLoader = classLoader;
    }
}
